package ar.edu.itba.pod.legajo48240.balance;

import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.sql.rowset.spi.SyncResolver;

import org.apache.log4j.Logger;

import ar.edu.itba.balance.api.AgentsBalancer;
import ar.edu.itba.balance.api.NodeAgent;
import ar.edu.itba.balance.api.NotCoordinatorException;
import ar.edu.itba.node.NodeInformation;
import ar.edu.itba.pod.legajo48240.model.Host;
import ar.edu.itba.pod.legajo48240.utils.Net;
import ar.edu.itba.pod.thread.CleanableThread;

public class Balancer extends UnicastRemoteObject implements AgentsBalancer {

	private static final long serialVersionUID = 1L;

	private final Host localhost;

	private final static Logger LOGGER = Logger.getLogger(Balancer.class);

	private final AtomicBoolean iWishToCoordinate = new AtomicBoolean();

	private final ConcurrentHashMap<NodeInformation, Long> elections = new ConcurrentHashMap<NodeInformation, Long>();

	private final ConcurrentHashMap<NodeInformation, Long> coordinator = new ConcurrentHashMap<NodeInformation, Long>();

	private final PriorityBlockingQueue<NodeRank> agentsOnNode = new PriorityBlockingQueue<NodeRank>(30,new CompareRank());
	
	private CountDownLatch inElection = new CountDownLatch(1);

	private final int AWAIT_FOR_COORDINATOR = 3000; // milliseconds

	public Balancer(Host localhost) throws RemoteException {
		this.localhost = localhost;
	}

	@Override
	public void bullyElection(NodeInformation node, long timestamp)
	throws RemoteException {

		if(!hasArrivedElection(node, timestamp)){
			
			if (node.id().compareTo(localhost.node().id()) == 0) {
				// ignore bullyElection
				startNewElection(node,timestamp);
				LOGGER.debug(" start election..."  );

			} else {

				// in election local host flag
				localhost.setCoordinator(null);

				iWishToCoordinate.set(true);
				inElection=new CountDownLatch(1);

				if (node.id().compareTo(localhost.node().id()) < 0) {
					LOGGER.debug("MENOR " + node.id() );
					try{
						AgentsBalancer remoteBalancer = Net.getBalancer(node);
						remoteBalancer.bullyOk(localhost.node());
						LOGGER.info(localhost.node().id() +" send Bully OK to " + node.id() );
					} catch( NotBoundException e){
						Net.disconnectFromGroup(localhost,node);
					}
					startNewElection(localhost.node(),System.nanoTime());


				} else if (node.id().compareTo(localhost.node().id()) > 0 ) {
					LOGGER.debug("MAYOR " + node.id());
					forwardBullyElection(node,timestamp, iWishToCoordinate);
				}
			}
		}
	}

	private void forwardBullyElection(final NodeInformation node, final long timestamp,
			final AtomicBoolean iWishToCoordinate) {

		CleanableThread forwardElection = new CleanableThread("bullyForward") {
			public void run() {
				super.run();

				// TODO: iwish ??

				//				while(iWishToCoordinate.get()){
				try {
					for(NodeInformation node: localhost.network().connectedNodes()){
						if(!node.equals(localhost.node())){
							try {
								AgentsBalancer remoteBalancer=Net.getBalancer(node);
								remoteBalancer.bullyElection(node, timestamp);
								LOGGER.debug(localhost.node().id() + " forward to " + node.id());
							} catch (NotBoundException e) {
								Net.disconnectFromGroup(localhost, node);
							}
						}
					}

					try {
						Thread.sleep(AWAIT_FOR_COORDINATOR);

					} catch (InterruptedException e) {} //ignore

					LOGGER.debug("stop waiting coordinator.... wish: " + iWishToCoordinate.get());

					if(iWishToCoordinate.get()){
						iAmCoordinator(node,timestamp);
					}
				} catch (RemoteException ignore) {} // local method
			}
		};
		forwardElection.start();

	}

	private void startNewElection(final NodeInformation localNode, final long now) {

		LOGGER.info(localhost.node().id() + " START NEW ELECTION");

		CleanableThread awaitCoordinator = new CleanableThread("awaitCoordinator") {

			@Override
			public void run() {
				super.run();
				try {
					try {
						for(NodeInformation node: localhost.network().connectedNodes()){
							if(!node.equals(localhost.node())){
								try {
									AgentsBalancer remoteBalancer=Net.getBalancer(node);
									remoteBalancer.bullyElection(localNode, now);
									LOGGER.debug(localhost.node().id() + " send election to " + node.id());
								} catch (NotBoundException e) {
									Net.disconnectFromGroup(localhost, node);
								}
							}
						}
					} catch (RemoteException ignore) {} // local method

					Thread.sleep(AWAIT_FOR_COORDINATOR);

					LOGGER.debug("stop waiting coordinator.... wish: " + iWishToCoordinate.get());

					if(iWishToCoordinate.get()){

						iAmCoordinator(localNode,System.nanoTime());
					}
				} catch (InterruptedException ignore) {	}
			}
		};

		awaitCoordinator.start();

	}

	protected void iAmCoordinator(NodeInformation localNode, long now) {

		try {
			for(NodeInformation node: localhost.network().connectedNodes()){
				// tell everyone i am coordinator, even me
				try {
					AgentsBalancer remoteBalancer=Net.getBalancer(node);
					remoteBalancer.bullyCoordinator(localNode, now);
					LOGGER.debug(localhost.node().id() + " say " + localNode+ " isCoordinator" );
				} catch (NotBoundException e) {
					try {
						localhost.network().disconnectFromGroup(node);
					} catch (NotBoundException ignore) {}
				}
			}
		} catch (RemoteException ignore) {} // local method
	}

	/**
	 * check if bully Election request was requested
	 * 
	 * @param node
	 * @param timestamp
	 * @return
	 */
	private boolean hasArrivedElection(NodeInformation node, long timestamp) {

		Long hasArrive=elections.get(node);
		
		if(hasArrive==null || hasArrive < timestamp){
			// if never arrive an election from the node or it is a new election
			elections.put(node, timestamp);
			LOGGER.debug("Arrive in " + localhost.node().id() + " new election for " + node.id());
			return false;
		}
		return true;
	}


	@Override
	public void bullyOk(NodeInformation node) throws RemoteException {
		LOGGER.debug("BullyOk received from: " + node.id());
		iWishToCoordinate.set(false);
	}

	@Override
	public void bullyCoordinator(NodeInformation node, long timestamp)
	throws RemoteException {

		Long alreadyArrived=coordinator.get(node);

//		LOGGER.info("bullyCoordinator");
		
		if(alreadyArrived==null || alreadyArrived < timestamp){

			coordinator.put(node, timestamp);
			
			LOGGER.info("in " + localhost.node().id() + " set cordinator: " + node.id() );
			
			localhost.setCoordinator(node);
			inElection.countDown();
			

			if(localhost.node().equals(node)){
				moveTowardTheNovice();
			}

			LOGGER.debug(node.id() + " say IS NEW COORDINATOR of " + localhost.node().id()  + " in " + localhost.network().getGroupId() + " new Coordinator : " + localhost.getCoordinator().id());

			for (NodeInformation n : localhost.network().connectedNodes()) {
				if (!n.equals(localhost.node())) {
					try {
						AgentsBalancer rBalancer = Net.getBalancer(node);
						rBalancer.bullyCoordinator(node, timestamp);
						LOGGER.debug(localhost.node().id() + " forward " + node.id()	+ " is new coordinator");
					} catch (NotBoundException e) {
						Net.disconnectFromGroup(localhost, n);
					}
				}
			}
		}

	}

	// TODO: synchronized

	private synchronized void moveTowardTheNovice() {
		
		try {
			inElection.await();
		} catch (InterruptedException e) {	e.printStackTrace();}


		LOGGER.info("" );
		LOGGER.info("MOVE TOWARD THE NOVICE----- from: " + localhost.node().id() );
		LOGGER.info("" );

		/**
		 *  CLEAR AGENTS NODE
		 */
		agentsOnNode.clear();

		// pido la cantidad de agentes de cada nodo de todo el cluster
		// N A B C 
		// 0 7 7 8 

		buildAgentsOfNodes();
		
		LOGGER.info("MTN: " + agentsOnNode + " " + agentsOnNode.size());

		if(agentsOnNode.size()>1){

			// armo la lista de nodos que deberia pasar al novato
			// balance agents
			//  5  5  6  6 
			// -2 -2 -2 +6

			List<NodeAgent> exilesNodes = new ArrayList<NodeAgent>();
			distribute(exilesNodes);
			
		}else{
			LOGGER.info("JUST ONE >> " + agentsOnNode);
		}

	}

	private void distribute(List<NodeAgent> exilesNodes) {
		
		AgentQty qty=AgentsOnNodeQty();
		
		double totalAgents=qty.totalAgents();
		double nodes=qty.nodes();

		
		int avg=(int)(totalAgents/nodes);      // ex. 5
		int rest= (int)(totalAgents % nodes);  // ex. 2
		
		LOGGER.info( "avg:"  + avg + " rest:" + rest);
		
		if(nodes==0){ LOGGER.error("NULL DIV !"); }
		
		List<Integer> distribution = new ArrayList<Integer>();
		
		if(!exilesNodes.isEmpty()){
			double exilesQty=exilesNodes.size();
			int oldNodes=agentsOnNode.size();
			
			int exilesAvg=(int)(exilesQty/oldNodes);
			int exilesMod=(int)(exilesQty%oldNodes);
			
			for(NodeRank nrnk:agentsOnNode){
				int dif = exilesAvg;
				if(--oldNodes<exilesMod)dif++;
				if(dif>0){
					LOGGER.info("add distribution: " + dif);
					distribution.add(dif);
				}
			}
		}
		
		for(NodeRank nrnk:agentsOnNode){
			int dif = avg-nrnk.agents();
			if(--nodes<rest)dif++;
			if(dif>0){
				LOGGER.info("add distribution: " + dif);
				distribution.add(dif);
			}else if (dif<0){
//				LOGGER.info("iteratin... on " + nrnk.node().id() + " agents "+ nrnk.agents() + " stopAndGet: "  + dif);
				try {
					exilesNodes.addAll(Net.getTransferer(nrnk.node()).stopAndGet(-dif));
					agentsOnNode.remove(nrnk);
					nrnk.setAgent(nrnk.agents()+dif);
					agentsOnNode.offer(nrnk);
				} catch (Exception e) {
					Net.disconnectFromGroup(localhost, nrnk.node());
				}
			}

		}
		
		
		if(exilesNodes.size()>0){
			
			int pointer=0;
			for(Integer part:distribution){
				NodeRank lowest=agentsOnNode.remove();
				NodeInformation novice=lowest.node();
				List<NodeAgent> l=new ArrayList<NodeAgent>();
				l.addAll(exilesNodes.subList(pointer, pointer+part));
				pointer=part;
				try {
					Net.getTransferer(novice).runAgentsOnNode(l);
					agentsOnNode.offer(new NodeRank(novice,lowest.agents()+l.size()));
				} catch (Exception e) {
					Net.disconnectFromGroup(localhost, novice);
				}
			}
			LOGGER.info("BALANCED >> " + agentsOnNode);
		}else{
			LOGGER.info("NO BALANCED NEEDED >> " + agentsOnNode);
		}

		
	}

	private AgentQty AgentsOnNodeQty() {
		
		double totalAgents=0;
		double nodes=0;
		for(NodeRank nrnk:agentsOnNode){
			totalAgents+=nrnk.agents();
			nodes++;
		}
		return new AgentQty(totalAgents,nodes);
	}


	private class AgentQty{
		private final double totalAgents;
		private final double nodes;

		public AgentQty(double totalAgents, double nodes) {
			this.totalAgents=totalAgents;
			this.nodes=nodes;
		}
		public double nodes(){
			return nodes;
		}
		public double totalAgents(){
			return totalAgents;
		}

	}
	private  void buildAgentsOfNodes() {
		try {
			for( NodeInformation node: localhost.network().connectedNodes()){
				// for everyOne even me 
				try {
					int agents=Net.getTransferer(node).getNumberOfAgents();
					agentsOnNode.add(new NodeRank(node, agents));
				} catch (Exception e) {
					Net.disconnectFromGroup(localhost, node);
				}
			}
		} catch (RemoteException ignore) {	} // local method

	}

	@Override
	public void shutdown(List<NodeAgent> agents) throws RemoteException,
	NotCoordinatorException {

		if( !localhost.getCoordinator().equals(localhost.node())){
			throw new NotCoordinatorException(localhost.getCoordinator());
		}
		
		
		// saco el nodo que se da de baja
		LOGGER.info("SHUTDOWN: node: " + agents.get(0).node().id() + " AON: " + agentsOnNode);
		
		agentsOnNode.remove(new NodeRank(agents.get(0).node(), 0));
		
		LOGGER.info("AON: " + agentsOnNode);
		
		if(agentsOnNode.size()>=1){

			distribute(agents);
			
		}else{
			LOGGER.info("SHUTDOWN LAST ONE");
		}
		

	}

	private boolean myContains(PriorityBlockingQueue<NodeRank> agentsOnNode,
			NodeInformation node) {
		
		for(NodeRank nr: agentsOnNode){
			if(nr.node().id().equals(node.id())){
				return true;
			}
		}
		return false;
	}
		
	@Override
	public void addAgentToCluster(NodeAgent agent) throws RemoteException,
	NotCoordinatorException {

		// get Coordination is a blocking method. return only if a cluster has a coordinator has
		if( !localhost.getCoordinator().equals(localhost.node())){
			LOGGER.info(localhost.node().id() +" i am not current coordinator :(");
			throw new NotCoordinatorException(localhost.getCoordinator());
		}

		LOGGER.info( "" );

		NodeInformation lowest;

		synchronized (Balancer.class) {


			if(myContains(agentsOnNode,agent.node())){
				
//			}
//			if(agentsOnNode.contains(new NodeRank(agent.node(), 0))){
				// tomo el menor
				NodeRank rank=agentsOnNode.remove();

				LOGGER.info("current novice:" + rank.node().id());

				agentsOnNode.add(new NodeRank(rank.node(), rank.agents()+1));
				lowest=rank.node();
				LOGGER.info( "upgrade lowest NodeRank: " + rank.node() + " a " + (rank.agents()+1));

			}else{

				// tomo el menor
				LOGGER.info( "new NodeRank entry in: " + agent.node() );
				agentsOnNode.add(new NodeRank(agent.node(), 1));
				lowest=agent.node();
			}

			for(NodeRank nr:agentsOnNode){
				LOGGER.info( "AON >> " +nr.node().id() + " agents: "+ nr.agents());
			}

			List<NodeAgent> nodeAgent=new ArrayList<NodeAgent>();
			nodeAgent.add(new NodeAgent(lowest, agent.agent()));

			LOGGER.info( localhost.node().id() + " says to " + lowest.id() + " addAgent " + agent.agent() );

			try {
				Net.getTransferer(lowest).runAgentsOnNode(nodeAgent);
			} catch (Exception e) {
				e.printStackTrace();
				Net.disconnectFromGroup(localhost,agent.node());
			}
		}

	}


}
